// Created: Dec 13, 2014
//
// Author: wjinwen.1988@gmail.com

#include "chi/net/event_loop.h"

#include <stdint.h>
#include <unistd.h>
#include <sys/eventfd.h>
#include <sys/socket.h>

#include <functional>
#include <algorithm>

#include <glog/logging.h>

#include "chi/net/channel.h"

namespace chi {
namespace net {

static void EventHandler(int fd, short what, void* arg) {
  EventLoop* loop = static_cast<EventLoop*>(arg);
  loop->HandleEvent(fd, what);
}

static void TimeoutHandler(int fd, short what, void* arg) {
  Timer* timer = static_cast<Timer*>(arg);
  timer->loop->Cancel(timer->timerid);
  timer->cb(timer->timerid);
}

static void TimerHandler(int fd, short what, void* arg) {
  Timer* timer = static_cast<Timer*>(arg);
  timer->cb(timer->timerid);
}

static int CreateEventfd() {
  int evfd = ::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
  if (evfd < 0) {
    LOG(ERROR) << "failed in eventfd";
    abort();
  }
  return evfd;
}

EventLoop::EventLoop() 
  : looping_(false),
    quit_(false),
    calling_pending_functors_(false),
    loop_thread_(0),
    base_(event_base_new()),
    wakeup_fd_(CreateEventfd()),
    wakeup_channel_(new Channel(this, wakeup_fd_)),
    timer_sequence_(0),
    mutex_() {
  wakeup_channel_->SetReadCallback(std::bind(&EventLoop::HandleRead, this));
  wakeup_channel_->EnableReading(); 
}

EventLoop::~EventLoop() {
  ::close(wakeup_fd_);
  if (wakeup_channel_) delete wakeup_channel_;

  for(const auto& it : channels_) {
    auto chpair = it.second;
    event* ev = chpair.second;
    event_del(ev);
    event_free(ev);
  }
  channels_.clear();

  for (const auto& it : timers_) {
    Timer* t = it.second;
    event_del(t->timer_ev);
    event_free(t->timer_ev);
  }
  
  if (base_) 
    event_base_free(base_);
}

void EventLoop::Loop() {
  assert(!looping_);
  looping_ = true;
  quit_ = false;

  loop_thread_.store(pthread_self(), std::memory_order_release);

  while (!quit_) {
    event_base_loop(base_, EVLOOP_ONCE);
    DoPendingFunctors();
  }

  loop_thread_.store(0, std::memory_order_release);
  looping_ = false;
}

void EventLoop::Quit() {
  quit_ = true;
  Wakeup();
}

void EventLoop::RunInLoop(Functor&& cb) {
  if (IsInEvenLoop()) {
    cb();
  } else {
    QueueInLoop(std::move(cb));
  }
}

void EventLoop::QueueInLoop(Functor&& cb) {
  {
    MutexLockGuard lock(mutex_);
    pending_functors_.push_back(std::move(cb));
  }
  Wakeup();
}

void EventLoop::RunAfter(double delay, TimeoutCallback&& cb) {
  Timer* timer = new Timer();
  timer->loop = this;
  timer->timer_ev = evtimer_new(base_, TimeoutHandler, timer);
  timer->tv.tv_sec = delay;
  timer->tv.tv_usec = (uint64_t)(delay * 1000) % 1000;
  timer->cb = std::move(cb);
  RunInLoop(std::bind(&EventLoop::AddTimer, this, timer));
}

void EventLoop::RunEvery(double interval, TimeoutCallback&& cb) {
  Timer* timer = new Timer();
  timer->loop = this;
  timer->timer_ev = event_new(base_, -1, EV_PERSIST, TimerHandler, timer);
  timer->tv.tv_sec = interval;
  timer->tv.tv_usec = (uint64_t)(interval * 1000) % 1000;
  timer->cb = std::move(cb);
  RunInLoop(std::bind(&EventLoop::AddTimer, this, timer));
}

void EventLoop::Cancel(uint64_t timerid) {
  RunInLoop(std::bind(&EventLoop::DelTimer, this, timerid));
}

void EventLoop::Wakeup() {
  uint64_t one = 1;
  ssize_t n = ::write(wakeup_fd_, &one, sizeof(one));
  if (n != sizeof(one)) {
    LOG(ERROR) << "writes " << n << " bytes instead of 8";
  }
}

void EventLoop::UpdateChannel(Channel* ch) {
  event* ev = nullptr;
  int sockfd = ch->fd();
  auto it = channels_.find(sockfd);
  if (it != channels_.end()) {
    ChannelPair& chpair = it->second;
    ev = chpair.second;
    assert(ev != nullptr);
    event_del(ev);
    if (ch->events() != 0) {
      event_assign(ev, base_, sockfd, ch->events(), EventHandler, this);
      event_add(ev, nullptr);
    } else {
      event_free(ev);
      channels_.erase(it);
    }
  } else {
    ev = event_new(base_, sockfd, ch->events(), EventHandler, this);
    event_add(ev, nullptr);
    channels_[sockfd] = std::make_pair(ch, ev); 
  }
}

void EventLoop::HandleEvent(int fd, short what) {
  auto it = channels_.find(fd);
  if (it != channels_.end()) {
    Channel* ch = it->second.first;
    ch->set_revents(what);
    ch->HandleEvent();
  } else {
    LOG(ERROR) << "handle unknown event, fd: "<< fd << " what: "<< what; 
  }
}

void EventLoop::AddTimer(Timer* timer) {
  assert(timer != nullptr);
  ++timer_sequence_;
  timer->timerid = timer_sequence_;
  evtimer_add(timer->timer_ev, &(timer->tv));
  timers_[timer->timerid] = timer;
  VLOG(11) << "add timer " << timer->timerid;
}

void EventLoop::DelTimer(uint64_t timerid) {
  auto it =  timers_.find(timerid);
  if (it != timers_.end()) {
    Timer* timer = it->second;
    evtimer_del(timer->timer_ev);
    event_free(timer->timer_ev);
    delete timer;
    timers_.erase(it);
    VLOG(11) << "delete timer " << timerid;
  } else {
    LOG(WARNING) << "not found the timer " << timerid;
  }
}

void EventLoop::HandleRead() {
  uint64_t one = 1;
  ssize_t n = ::read(wakeup_fd_, &one, sizeof(one));
  if (n != sizeof(one)) {
    LOG(ERROR) << "reads " << n << " bytes instead of 8";
  }
  
  DoPendingFunctors();
}

void EventLoop::DoPendingFunctors() {
  calling_pending_functors_ = true;
  std::vector<Functor> functors;
  {
    MutexLockGuard lock(mutex_);
    functors.swap(pending_functors_);
  }

  for (size_t i = 0; i < functors.size(); ++i) {
    functors[i]();
  }
  calling_pending_functors_ = false;
}

} 
}
